package com.bff.gaia.streaming.api.xjoin.client.hbase.parse;

import com.bff.gaia.streaming.api.xjoin.client.hbase.core.table.HbaseSideTableInfo;
import com.bff.gaia.streaming.api.xjoin.core.utils.ClassUtil;
import com.bff.gaia.streaming.api.xjoin.core.utils.TypeInformationUtil;
import sun.rmi.runtime.Log;

import java.util.Arrays;
import java.util.regex.Pattern;

public class HbaseSideTableInfoBuilder {
	private final static String FIELD_KEY = "fieldKey";

	private final static Pattern FIELD_PATTERN = Pattern.compile("(?i)(.*)\\s+AS\\s+(\\w+)$");

	public static final String HBASE_ZOOKEEPER_QUORUM = "zookeeperQuorum";

	public static final String ZOOKEEPER_PARENT = "zookeeperParent";

	public static final String TABLE_NAME_KEY = "tableName";

	public static final String PRE_ROW_KEY = "preRowKey";

	public static final String CACHE = "cache";

	public static TableInfoBuilder buildHbaseSideTableInfo() {
		return new TableInfoBuilder();
	}

	public static class TableInfoBuilder {

		private HbaseSideTableInfo hbaseTableInfo;

		TableInfoBuilder() {
			this.hbaseTableInfo = new HbaseSideTableInfo();
			this.hbaseTableInfo.setPreRowKey(false);
		}

		public TableInfoBuilder setTableName(String tableName) {
			this.hbaseTableInfo.setName(tableName.toLowerCase());
			this.hbaseTableInfo.setTableName(tableName.toLowerCase());
			return this;
		}

		public TableInfoBuilder setHost(String host) {
			this.hbaseTableInfo.setHost(host.toLowerCase());
			return this;
		}

		public TableInfoBuilder setParent(String parent) {
			this.hbaseTableInfo.setFamily(parent.toLowerCase());
			return this;
		}

		public TableInfoBuilder setPrimaryKeys(String... primaryKeys) {
			this.hbaseTableInfo.setPrimaryKeys(Arrays.asList(primaryKeys));
			return this;
		}

		public TableInfoBuilder addfieldInfo(String fieldName, String fieldType) {
			hbaseTableInfo.addField(fieldName);
			hbaseTableInfo.addFieldType(fieldType);
			hbaseTableInfo.addFieldClass(ClassUtil.stringConvertClass(fieldType));
			hbaseTableInfo.addFieldTypeInfomation(TypeInformationUtil.stringConvertTypeInformation(fieldType));
			return this;
		}

		public TableInfoBuilder enableQualifierQuery(){
			hbaseTableInfo.setQualifierQuery(true);
			return this;
		}

		public TableInfoBuilder addfieldInfos(String[] fields) {
			//TODO make sure do not exist type error
			for (String str : fields) {
				String fieldName = str.split(" ")[0];
				String fieldType = str.split(" ")[1];
				hbaseTableInfo.addField(fieldName);
				hbaseTableInfo.addFieldType(fieldType);
				hbaseTableInfo.addFieldClass(ClassUtil.stringConvertClass(fieldType));
				hbaseTableInfo.addFieldTypeInfomation(TypeInformationUtil.stringConvertTypeInformation(fieldType));
			}
			return this;
		}

		public HbaseSideTableInfo finish() {
			if(!hbaseTableInfo.isQualifierQuery()){
				if (hbaseTableInfo.getPrimaryKeys() == null) {
					this.hbaseTableInfo.setPrimaryKeys(hbaseTableInfo.getFieldList());
				}
			}

			for (String s : hbaseTableInfo.getFieldList()) {
				hbaseTableInfo.addColumnRealName(hbaseTableInfo.getFamily() + ":" + s);
			}
			hbaseTableInfo.finish();
			return this.hbaseTableInfo;
		}
	}
}